package org.example

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object SparkStreamingTY {
  def main(args: Array[String]): Unit = {
    println("进入套接函数")
    val myIp = "192.168.72.130"
    val port = "9000" // 确保端口是字符串形式，并且是一个有效的端口号

    // 创建 Spark 配置
    val conf = new SparkConf().setAppName("ty03").setMaster("local[*]")

    // 创建 StreamingContext
    val ssc = new StreamingContext(conf, Seconds(1))

    // 从套接字读取数据流
    val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream(myIp, port.toInt, StorageLevel.MEMORY_AND_DISK_SER)

    // 处理数据流
    val wordCounts: DStream[(String, Int)] = socketDS
      .flatMap(_.split(" ")) // 将每一行拆分成单词
      .map((_, 1)) // 将每个单词映射为 (word, 1)
      .reduceByKey(_ + _) // 对相同 key 的值进行累加

    // 打印结果
    wordCounts.print()

    // 启动 StreamingContext 并等待终止
    ssc.start()
    ssc.awaitTermination()
  }
}
